package com.atguigu.tingshu.album.receiver;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.service.AlbumMqService;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.rabbit.constant.MqConst;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.user.UserListenProcessVo;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * ClassName: AlbumReceiver
 * Package: com.atguigu.tingshu.album.receiver
 * Description:
 *
 * @Author 梅超凡
 * @Create 2025/2/5 20:38
 * @Version 1.0
 */
@Component
@Slf4j
public class AlbumReceiver {



    @Autowired
    private AlbumMqService albumMqService;

    @Autowired
    private StringRedisTemplate redisTemplate;


    /**
     * 消费声音转台更新消息
     * 对于队列 交换机 以及绑定关系可以通过配置来指定 也可以通过下游rabbitlistern注解指定
     */

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = MqConst.QUEUE_TRACK_STAT_UPDATE, durable = "true", exclusive = "false"),
            exchange = @Exchange(value = MqConst.EXCHANGE_TRACK),
            key = MqConst.ROUTING_TRACK_STAT_UPDATE))
    @SneakyThrows   // 可以绕开编译期间的异常 等运行时发现异常才抛出
    public void listenAlbumAndTrackStatUpdate(String content, Message message, Channel channel) {

        // 1.获取消息，进行判断
        if (StringUtils.isEmpty(content)) {
            return;
        }

        //2.消息转换
        UserListenProcessVo userListenProcessVo = JSONObject.parseObject(content, UserListenProcessVo.class);
        TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
        trackStatMqVo.setAlbumId(userListenProcessVo.getAlbumId());
        trackStatMqVo.setTrackId(userListenProcessVo.getTrackId());
        trackStatMqVo.setBusinessNo(UUID.randomUUID().toString().replaceAll("-", ""));
        trackStatMqVo.setStatType(SystemConstant.TRACK_STAT_PLAY);
        trackStatMqVo.setCount(1);

        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 2.消费消息
        try {
            albumMqService.listenAlbumAndTrackStatUpdate(trackStatMqVo);
            // 4.手动应答
            channel.basicAck(deliveryTag, false);
        } catch (GuiguException e) {
            //重试 消息能够继续被消费
            String retryMsgKey = "retry:msg:" + trackStatMqVo.getBusinessNo();

            Long count = redisTemplate.opsForValue().increment(retryMsgKey);
            if (count == null) {
                count = 0L; // 如果 count 为 null，则初始化为 0
            }
            if (count >= 3L) {
                log.error("消息重试了{}次,异常原因：{}，请人工排查", count, e.getMessage());
                channel.basicNack(deliveryTag,false,false);
            } else {
                log.error("消息重试了{}次,异常原因：{}，正在重试", count, e.getMessage());
                channel.basicNack(deliveryTag,false,true);
            }
        }catch (Exception e){
            log.error("网络错误，消费消息失败：{}", e.getMessage());
            channel.basicNack(deliveryTag,false,false);
        }
    }


    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(
                            value = MqConst.QUEUE_ALBUM_STAT_UPDATE,
                            durable = "true",
                            exclusive = "false"),
            exchange = @Exchange(value = MqConst.EXCHANGE_ALBUM),
                    key = MqConst.ROUTING_ALBUM_STAT_UPDATE))
    @SneakyThrows   // 可以绕开编译期间的异常 等运行时发现异常才抛出
    public void listenAlbumAndBuyNumUpdate(String orderNo, Message message, Channel channel) {
        // 1.获取消息，进行判断
        if (StringUtils.isEmpty(orderNo)) {
            return;
        }
        // 2.消费消息
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 调用服务更新专辑的购买数量统计
            albumMqService.updateAlbumAndBuyNum(orderNo);
            // 3.手动应答
            //1. ACK（确认消息）
            //作用：当消费者成功处理完一条消息后，向RabbitMQ发送一个确认信号，表示该消息已经被成功消费。RabbitMQ接收到这个确认信号后，会将该消息从队列中移除。
            //使用场景：
            //消费者成功处理了消息，并且希望确保该消息不会被再次投递。
            //确保消息的可靠性传递，防止消息丢失。
            channel.basicAck(deliveryTag, false);
        } catch (GuiguException e) {
            // 重试 消息继续能够被消费者消费
            String retryMsgKey = "retry:msg:" + orderNo;
            Long count = redisTemplate.opsForValue().increment(retryMsgKey);
            if (count == null) {
                count = 0L; // 如果 count 为 null，则初始化为 0
            }
            if (count >= 3L) {
                // 如果重试次数超过3次，则记录错误日志并拒绝消息
                log.error("消息重试了{}次，异常原因：{}，请人工排查", count, e.getMessage());
                //2. NACK（拒绝消息）
                //作用：当消费者无法成功处理一条消息时，向RabbitMQ发送一个拒绝信号。根据参数的不同，可以选择重新入队该消息或直接丢弃。
                //使用场景：
                //消费者处理消息失败，但希望消息能够被重新消费（例如重试机制）。
                //消费者处理消息失败，并且确定不需要再处理该消息（例如超过重试次数）。
                channel.basicNack(deliveryTag, false, false);  // 删除掉
            } else {
                log.error("消息重试了{}次", count);
                channel.basicNack(deliveryTag, false, true);
            }
        } catch (Exception e) {
            log.error("手动应答期间，网络出现了抖动，但是业务已经被成功执行过，消息直接丢掉");
            channel.basicNack(deliveryTag, false, false);
        }
    }
}
